package cn.monkey.transport.mq.rabbitmq;

import cn.monkey.transport.mq.rabbitmq.codec.Codec;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.rabbitmq.OutboundMessage;
import reactor.rabbitmq.OutboundMessageResult;
import reactor.rabbitmq.SendOptions;
import reactor.rabbitmq.Sender;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.stream.Collectors;

public class SimpleRabbitMqOutbound implements RabbitMqOutbound {

    protected final Sender sender;
    protected final Codec codec;
    protected final String exchange;
    protected final String routerKey;
    protected final SendOptions sendOptions;
    protected final BlockingQueue<Object> cacheQueue;
    protected final int cacheSize;

    public SimpleRabbitMqOutbound(Sender sender,
                                  Codec codec,
                                  String exchange,
                                  String routerKey,
                                  SendOptions sendOptions,
                                  int cacheSize) {
        this.sender = sender;
        this.exchange = exchange;
        this.routerKey = routerKey;
        this.sendOptions = sendOptions;
        this.cacheSize = cacheSize;
        this.cacheQueue = new ArrayBlockingQueue<>(cacheSize);
        this.codec = codec;
    }

    @Override
    public Mono<Void> write(Object o) {
        boolean offer = this.cacheQueue.offer(o);
        return offer ? Mono.empty() : Mono.error(new UnsupportedOperationException("cache is full"));
    }

    @Override
    public Mono<Void> writeAndFlush(Object o) {
        Flux<OutboundMessageResult<OutboundMessage>> resultFlux = this.sender.sendWithTypedPublishConfirms(subscriber ->
                subscriber.onNext(new OutboundMessage(this.exchange, this.routerKey, this.codec().encode(o))), this.sendOptions);
        return resultFlux.then();
    }

    @Override
    public Mono<Void> flush() {
        if (this.cacheQueue.isEmpty()) {
            return Mono.empty();
        }
        ArrayList<Object> arrayList = new ArrayList<>(this.cacheQueue.size());
        this.cacheQueue.drainTo(arrayList);
        List<Flux<OutboundMessageResult<OutboundMessage>>> fluxList = arrayList.stream()
                .map(o -> this.sender.sendWithTypedPublishConfirms(subscriber -> subscriber.onNext(new OutboundMessage(this.exchange, this.routerKey, this.codec().encode(o))), this.sendOptions))
                .collect(Collectors.toList());
        return Flux.merge(fluxList).then();
    }

    @Override
    public String topic() {
        return this.exchange;
    }

    @Override
    public Mono<Void> close() {
        this.sender.close();
        return Mono.empty();
    }

    @Override
    public Codec codec() {
        return this.codec;
    }

    @Override
    public int cacheSize() {
        return this.cacheSize;
    }

    @Override
    public String exchange() {
        return this.exchange;
    }

    @Override
    public String routerKey() {
        return this.routerKey;
    }

    @Override
    public SendOptions sendOptions() {
        return sendOptions;
    }
}
